package org.idea.qiyu.framework.canal.consumer;

import com.alibaba.otter.canal.protocol.Message;
import org.idea.qiyu.framework.canal.handle.AbstractCanalHandler;
import org.idea.qiyu.framework.canal.handle.CanalMessageQueueHandler;
import org.idea.qiyu.framework.canal.producer.CanalMessageProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.HashSet;

/**
 * @Author linhao
 * @Date created in 2:32 下午 2021/6/20
 */
public class CanalMessageConsumer implements Runnable {

    private static final Logger LOGGER = LoggerFactory.getLogger(CanalMessageConsumer.class);

    private CanalMessageQueueHandler canalMessageQueueHandler;

    private HashSet<AbstractCanalHandler> abstractCanalHandlerList = new HashSet<>();

    public CanalMessageConsumer(CanalMessageQueueHandler canalMessageQueueHandler) {
        this.canalMessageQueueHandler = canalMessageQueueHandler;
    }

    public void addConsume(AbstractCanalHandler abstractCanalHandler) {
        this.abstractCanalHandlerList.add(abstractCanalHandler);
    }

    /**
     * 当接收到canal的消息之后就会触发这个批量开始消费的函数
     */
    public void startConsume(Message message) {
        if (CollectionUtils.isEmpty(abstractCanalHandlerList)) {
            return;
        }
        if (message.getId() < 0 || message.getEntries().size() == 0) {
            return;
        }
        for (AbstractCanalHandler abstractCanalHandler : abstractCanalHandlerList) {
            abstractCanalHandler.consume(message);
        }
    }

    @Override
    public void run() {
        LOGGER.info("[CanalMessageConsumer] is started!");
        while (true) {
            Message message = canalMessageQueueHandler.takeMessage();
            //单线程处理逻辑
            if (message != null) {
                startConsume(message);
            }
        }
    }
}
